package com.gongj.rabbit.produce.confirm;

import com.gongj.rabbit.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发送方确认机制
 */
public class confirm1 {

    public static void main11(String[] args) throws Exception {
        Connection conn = RabbitMQUtil.createConn();
        Channel channel = conn.createChannel();
        String quequ = "queue-2";
        String exchange = "exchange-55";
        String key = "key-22121";
        //创建交换机
        //channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
        //创建队列
       // channel.queueDeclare(quequ, true, false, false, null);
        //队列与交换机绑定
       // channel.queueBind(quequ, exchange, key);
        //将信道置为 publisher confirm 模式
        channel.confirmSelect();
        String message = "发送路由key为 = "+ key + "的消息";
        channel.basicPublish(exchange,key,null,message.getBytes());
        boolean b = channel.waitForConfirms();
        System.out.println("发送成功 = " + b);
    }

    public static void main2(String[] args) throws Exception {
        Connection conn = RabbitMQUtil.createConn();
        Channel channel = conn.createChannel();
        String quequ = "queue-2";
        String exchange = "exchange-242";
        String key = "key-2";
        //创建交换机
       // channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
        //创建队列
      //  channel.queueDeclare(quequ, true, false, false, null);
        //队列与交换机绑定
       // channel.queueBind(quequ, exchange, key);
        //将信道置为 publisher confirm 模式
        channel.confirmSelect();
        for (int i = 1; i < 10; i++) {
            String message = "发送路由key为 = "+ key + "的消息";
            channel.basicPublish(exchange,key,null,message.getBytes());
            boolean b = channel.waitForConfirms();
            System.out.println("发送成功" + b);
        }
    }
    //批量confirm
    public static void main3(String[] args) throws Exception {
        Connection conn = RabbitMQUtil.createConn();
        Channel channel = conn.createChannel();
        String quequ = "queue-2";
        String exchange = "exchange-222";
        String key = "key-2";
        //创建交换机
        //channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
        //创建队列
      //  channel.queueDeclare(quequ, true, false, false, null);
        //队列与交换机绑定
       // channel.queueBind(quequ, exchange, key);
        //将信道置为 publisher confirm 模式
        channel.confirmSelect();
        for (int i = 1; i < 10; i++) {
            String message = "发送路由key为 = "+ key + "的消息";
            channel.basicPublish(exchange,key,null,message.getBytes());
        }
        //批量确认信息，发送的消息中，如果有一条是失败的，则所有消息发送都会失败
        boolean b = channel.waitForConfirms();
        System.out.println("发送成功" + b);
    }


    //异步confirm
    public static void main(String[] args) throws IOException {
        Connection conn = RabbitMQUtil.createConn();
        Channel channel = conn.createChannel();
        String quequ = "queue-2";
        String exchange = "exchange-9527";
        String key = "key-2";
        //创建交换机
       // channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true);
        //创建队列
       // channel.queueDeclare(quequ, true, false, false, null);
        //队列与交换机绑定
      //  channel.queueBind(quequ, exchange, key);
        channel.confirmSelect();
        final ConcurrentSkipListMap<Long,String> map = new ConcurrentSkipListMap();
        // 添加一个异步确认的监听器
        channel.addConfirmListener(new ConfirmListener() {
            //参数一：deliveryTag： 消息的编号
            //参数二：multiple：是否批量confirm true 是
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("map 数据：" + map.size());
                if (multiple) {
                    //如果是批量确认  返回的是小于等于当前序列号的消息 是一个 map
                    ConcurrentNavigableMap<Long, String> confirmed =
                            map.headMap(deliveryTag, true);
                    //清除该部分未确认消息
                    confirmed.clear();
                    System.out.println("批量确认清楚 map 数据：" + map.size());
                }else{
                    //只清除当前序列号的消息
                    map.remove(deliveryTag);
                    System.out.println("只清除当前序列号的消息 map 数据：" + map.size());
                }
                System.out.println("消息发送到交换机成功,deliveryTag: " + deliveryTag + ", multiple: " + multiple);
            }

            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息发送到交换机失败, deliveryTag: " + deliveryTag + ", multiple: " + multiple);
                String message = map.get(deliveryTag);
                System.out.println("消息发送到交换机失败，发布的消息："+message+"未被确认，序列号为："+deliveryTag);
                //拿到了未确认的信息，可以进行其他逻辑，比如添加处理消息重发
            }
        });

        for (int i = 1; i < 6; i++) {
            String message = "发送路由key为 = "+ key + "的消息";
            // channel.getNextPublishSeqNo()获取下一个消息的序列号
            map.put(channel.getNextPublishSeqNo(),message);
            channel.basicPublish(exchange,key,null,message.getBytes());

        }
        System.out.println("其他逻辑");
    }

}

